package com.flc.gateway.mq;

import com.alibaba.fastjson.JSON;
import com.flc.gateway.Gateway;
import com.flc.gateway.config.AppConfig;
import com.flc.gateway.model.CommandHelper;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author haizi
 * @date 2018/1/23
 */
public class MqConsumer {

    private static final Logger logger = LoggerFactory.getLogger(MqConsumer.class);

    public static void start(AppConfig.RocketMqCfg rocketMqCfg) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CmdConsumer");

        consumer.setNamesrvAddr(rocketMqCfg.getNameServerAddr());
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        consumer.subscribe(rocketMqCfg.getCmdTopic(), AppConfig.getInstance().getServerCfg().getID());

        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            // 每次只能收到一条消息，并不是批量的
            String jsonStr = new String(msgs.get(0).getBody());
            Gateway.cmdQueue.put(JSON.parseObject(jsonStr, CommandHelper.class));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        consumer.start();

        logger.info("Consumer Started!!!");
    }
}
